//
// Created by 王筱童 on 2022/5/15.
//

#include"redis_server.h"
#include "senti_manager.h"
#include "others.h"
#include <iostream>

asio::awaitable<void> senti_manager::start(asio::ip::tcp::socket &&socket) {
    master=std::make_unique<asio::ip::tcp::socket>(std::move(socket));
    for(;;){
        bool tick_false= false;
        try {

            asio::steady_timer timer(server->timer_context,std::chrono::seconds(1));
            co_await timer.async_wait(asio::use_awaitable);
#ifdef TEST
            std::cout<<"Tick begin"<<std::endl;
#endif
            std::string command="PSENTI Tick";
            command= serialize(command);
            co_await asio::async_write(*master,asio::buffer(command),asio::use_awaitable);
            std::string head(sizeof(std::uint64_t),'\0');
            co_await asio::async_read(*master,asio::buffer(head),asio::use_awaitable);
            auto [len,_]=deserialize_head(head);
            head.resize(len,'\0');
            co_await asio::async_read(*master,asio::buffer(head),asio::use_awaitable);
#ifdef TEST
            std::cout<<"Tick end: "<<head<<std::endl;
#endif
            auto vec_str= split(head,'\n');
            decltype(slaves) tmp_slave;
            for(auto &str:vec_str){
                auto strs= split(str,' ');
                std::pair<std::string,std::uint16_t> pair=std::make_pair(strs[0], convert<std::uint16_t>(strs[1]));
                tmp_slave.push_back(pair);
            }
            slaves=tmp_slave;
        }
        catch (const std::exception&ec){
            std::cout<<"Tick Error: "<<ec.what()<<std::endl;
            tick_false= true;
        }
        if(tick_false){
            std::cout<<"Ticket Begin"<<std::endl;
            std::vector<int> tickets(slaves.size(),'0');
            std::vector<std::unique_ptr<asio::ip::tcp::socket>> sockets;
            for(auto& ip_port:slaves){
                std::string ip=ip_port.first;
                std::uint16_t port=ip_port.second;
                asio::ip::tcp::socket streamSocket(server->io_context);
                co_await streamSocket.async_connect({asio::ip::make_address(ip), port}, asio::use_awaitable);
#ifdef TEST
                std::cout<<"Ticket one ip_port: "<<ip<<":"<<std::to_string(port)<<std::endl;
#endif
                std::string data="PSENTI Ticket";
                data=data+" "+std::to_string(slaves.size());
                data= serialize(data);
                co_await asio::async_write(streamSocket,asio::buffer(data),asio::use_awaitable);
                std::string head(sizeof(std::uint64_t),'\0');
                co_await asio::async_read(streamSocket,asio::buffer(head),asio::use_awaitable);
                auto [len,_]=deserialize_head(head);
                head.resize(len,'\0');
                co_await asio::async_read(streamSocket,asio::buffer(head),asio::use_awaitable);
                std::uint64_t index= convert<std::uint64_t>(head);
#ifdef TEST
                std::cout<<"Ticket one res: "<<index<<std::endl;
#endif
                index=std::min(slaves.size()-1,index);
                tickets[index]++;
                sockets.emplace_back(std::make_unique<decltype(streamSocket)>(std::move(streamSocket)));
            }
            std::uint64_t res=std::max_element(tickets.begin(),tickets.end())-tickets.begin();
            master=std::move(sockets[res]);
            std::string command="SAVEOF "+slaves[res].first+" "+std::to_string(slaves[res].second);
            command= serialize(command);
            for(int i=0;i< sockets.size();i++){
                if(i==res){
                    continue;
                }
                co_await asio::async_write(*sockets[i],asio::buffer(command),asio::use_awaitable);
                std::string head(sizeof(std::uint64_t),'\0');
                co_await asio::async_read(*sockets[i],asio::buffer(head),asio::use_awaitable);
                auto [len,_]=deserialize_head(head);
                head.resize(len,'\0');
                co_await asio::async_read(*sockets[i],asio::buffer(head),asio::use_awaitable);
            }
#ifdef TEST
            std::cout<<"Ticket End"<<std::endl;
#endif
            tick_false= false;
        }
    }
}

senti_manager::senti_manager(redis_server *_server):server(_server) {

}
